#
# Reimplementation of the distributed testing infrastructure that was
# all hacked Tcl in the DTN stack.
#
import glob
import errno
import fcntl
import logging
import os
import re
import subprocess
import select
import shutil
import signal
import string
import sys
import threading
import time

import common

log = logging.getLogger('dtest')

#----------------------------------------------------------------------------
def set_ifn(d, k, v):
    """Set d[k] = v iff k isn't already in d"""
    if k in d:
        return
    d[k] = v

#----------------------------------------------------------------------------
class ConfigError(Exception):
    """Error raised by misconfiguration"""

#----------------------------------------------------------------------------
class Timeout(Exception):
    """Timeout occured waiting for a command to be executed"""
    
#----------------------------------------------------------------------------
class AppState:
    """Data structure for dealing with the state tree that
    applications use to describe their state changes to dtest

    applications generate colon statements seperated statements:

        tx:channel:tcp:node-1:handshake

    This translates to setting the key 'tx:channel:tcp:node-1' to
    the value 'handshake'. Note that another statement such as:

        tx:channel:tcp:node-1:connected

    Will overwrite the value stored at
    'tx:channel:tcp:node-1'. However, storing another
    tx:channel:udp:... will not.
    """
    def __init__(self):
        self.state = {}

    def set(self, stmt):
        s = stmt.strip().split(':')
        value = s[-1:][0]
        key = string.join(s[:-1], ':')
        self.state[key] = value

    def get_key(self, *args):
        """Matches keys of the form args[0]:args[1]:...

        If '*' is specified, matches any value.

	If the argument begins with '~', then the rest of the argument is
	treated as a regular expression.
	"""
        matches = []
        for k in self.state.keys():
            s = k.split(':')
            match = True
            for i in xrange(min(len(args), len(s))):
                if args[i] == '*':
                    continue
		if args[i][0] == '~':
		    r = re.compile(args[i][1:])
		    if not r.match(s[i]):
			match = False
                if s[i] != args[i]:
                    match = False
            if match:
                matches.append(k)
        return matches

    def get(self, *args):
        k = self.get_key(*args)
        assert len(k) <= 1
        if len(k) == 0:
            return None
        return self.state[k[0]]

#----------------------------------------------------------------------------
class DTest:
    """The main classes. Contains all of the other components (the
    mapper, hosts, nodes)"""

    IDLE_TIME = 0.1                     # only idle at most for 0.1 seconds
    
    def __init__(self, options = None, run_options = None):
        self.lock = threading.Lock()    # lock shared with the debugging thread
        self.hosts = {}
        self.nodes = {}
        self.mapper = None
        self.logger = Logger()
        self.runner = Runner(self)
        self.quit = False

        # This is the current state the test run is in. All timers
        # (e.g. 'at' events) are run offset by the event state. 
        self.event_state = 'start'
        self.event_base_time = 0
        self.events = {}                # state -> event sorted by time

        # State change callbacks, run when there is a state change
        # signaled from the apps
        self.state_change_cbs_ = []

        # options for general DTest related behavior
        #     debug       Output debug messages for DTest
        #     no_cleanup  Don't cleanup any of the files
        if options is None:
            options = []
        self.options = options

        # run_options are merged with each run_X
        #     debug       Run the task with debug output enabled
        if run_options is None:
            run_options = []
        self.run_options = run_options

        # default basedir for where the files will be located
        self.basedir_ = '/tmp/run'

        # parse user passed arguments
        self.parse_argv()

        if 'debug' in self.options:
            logging.basicConfig(level=logging.DEBUG)
        else:
            logging.basicConfig(level=logging.INFO)

    def host(self, name):
        return self.hosts[name]

    def node(self, name):
        return self.nodes[name]
            
    def parse_argv(self):
        for i in sys.argv[1:]:
            if i == '--debug-dtest':
                self.options.append('debug')
            if i == '--debug':
                self.run_options.append('debug')
            elif i == '--no-cleanup':
                self.options.append('no_cleanup')
            elif i.find('--basedir=') == 0:
                _, self.basedir_ = i.split('=')

    def add_host(self, host):
        self.hosts[host.name()] = host
        host.dtest_ = self

    def add_node(self, node):
        self.nodes[node.name()] = node
        node.dtest_ = self
        return node

    def set_mapper(self, mapper):
        self.mapper = mapper
        mapper.dtest_ = self
        if mapper.basedir_ is None:
            mapper.basedir_ = self.basedir_

    def reg_handler(self, handler):
        assert isinstance(handler, ProcAsyncHandler)
        handler.dtest_ = self
        self.runner.add_handler(handler)
        return handler

    def configure(self, skip = None):
        """Finish the configuration"""
        if skip is None:
            skip = []
        if 'mapping' not in skip:
            assert self.mapper
            self.mapper.map(self.nodes.values())
        if 'logging' not in skip:
            for n in self.nodes.values():
                n.setup_logging()
                
        log.info('Options: %s' % self.options)
        log.info('Run options: %s' % self.options)
                
    def setup_files(self):
        for n in self.nodes.values():
            n.setup_files()

    def run(self, f, *args):
        self.set_event_state('start')

        test_thread = TestRunner(self, f, *args)
        test_thread.start()
        
        try:
            while True:
                self.lock.acquire()
                try:
                    self.run_events()
                    timeout = self.next_event_time()
                    log.debug('next timeout = %s' % timeout)
                finally:
                    self.lock.release()
                    
                self.runner.wait_for_event(timeout)

                try:
                    self.lock.acquire()
                    self.check_state_change_cbs()
                    if self.quit:
                        log.info('Quit set, exiting...')
                        break
                finally:
                    self.lock.release()
        except KeyboardInterrupt, e:
            log.info('Got keyboard int, cleaning up...')
        except Exception, e:
            log.info('Got exception "%s", cleaning up...' % e)
            self.quit_cleanup()
            raise
        
        self.quit_cleanup()

    def quit_cleanup(self):
        """This allows the user to try a nice cleanup and when that
        fails, do a hard cleanup."""
        self.quit = True
        try:
            self.cleanup()
        except KeyboardInterrupt:
            log.info('Got keyboard int, killing immediately...')
            self.hard_cleanup()

    def cleanup(self):
        """Cleanup test run state"""
        if 'no_cleanup' in self.options:
            return
        self.runner.cleanup()
        for n in self.nodes.values():
            n.cleanup()

    def hard_cleanup(self):
        """Kill all running subprocesses immediately"""
        if 'no_cleanup' in self.options:
            return
        self.runner.hard_cleanup()
        for n in self.nodes.values():
            n.cleanup()        

    # Test running commands
    def check_state_change_cbs(self):
        to_rm = []
        for cb in self.state_change_cbs_:
            if not cb():
                to_rm.append(cb)
        for cb in to_rm:
            self.state_change_cbs_.remove(cb)
                
    def watch(self, cb):
        """Add a watch for state changes"""
        self.state_change_cbs_.append(cb)
    
    def set_event_state(self, state):
        log.debug('event state set to %s' % state)
        self.event_state = state
        self.event_base_time = time.time()
    
    def next_event_time(self):
        delta = time.time() - self.event_base_time
        assert delta >= 0.0
        if (self.event_state not in self.events or
            len(self.events[self.event_state]) == 0):
            return DTest.IDLE_TIME
        return min(DTest.IDLE_TIME,
                   self.events[self.event_state][0].seconds - delta)

    def run_events(self):
        delta = time.time() - self.event_base_time
        while (self.event_state in self.events and
               (len(self.events[self.event_state]) > 0 and
               self.events[self.event_state][0].seconds < delta)):
            e = self.events[self.event_state][0]
            self.events[self.event_state] = self.events[self.event_state][1:]
            log.debug('run event %s, %d events left' %
                      (e, len(self.events[self.event_state])))
            e.run_thunk()
    
    def add_event(self, origin, event):
        if origin in self.events:
            self.events[origin].append(event)
            self.events[origin].sort()
        else:
            self.events[origin] = [event]
        log.debug('added event %s:%s' % (origin, event))

    def at(self, origin, seconds, *pycmds):
        """Run a command at a certain time in the test run
        pycmds are strings containing python code (this is for
        easy indenting)"""
        e = Event(seconds, string.join(list(pycmds), ';'))
        self.add_event(origin, e)

    def at_f(self, origin, seconds, fcn):
        """Run a command at a certain time in the test run
        fcn is a python function"""
        e = Event(seconds, fcn)
        self.add_event(origin, e)
        
    def set_quit(self):
        self.quit = True

#----------------------------------------------------------------------------
class Event:
    def __init__(self, seconds, run):
        self.seconds = seconds
        self.run = run

    def __repr__(self):
        return '<dtest.Event %d %s>' % (self.seconds, self.run)

    def __cmp__(self, other):
        return cmp(self.seconds, other.seconds)
        
    def run_thunk(self):
        # XXX/bowei -- annoying issues with scope of variables that
        # would ideally be accessible.
        if isinstance(self.run, str):
            exec(self.run)
        else:
            self.run()

#----------------------------------------------------------------------------
class Runner:
    def __init__(self, dtest):
        self.handlers_ = []
        self.dtest = dtest

    def add_handler(self, h):
        assert not (h in self.handlers_)
        self.handlers_.append(h)
        
    def wait_for_event(self, timeout = None):
        # XXX/bowei watch for errors?
        self.dtest.lock.acquire()
        try:
            r = []
            r.extend([h.stdout() for h in self.handlers_ if h.watch_stdout()])
            r.extend([h.stderr() for h in self.handlers_ if h.watch_stderr()])
        finally:
            self.dtest.lock.release()
            
        if len(r) > 0:
            r, _, _ = select.select(r, (), (), timeout)
            self.dtest.lock.acquire()
            try:
                for i in r:
                    i.process()
            finally:
                self.dtest.lock.release()

        self.dtest.lock.acquire()
        try:
            for h in self.handlers_:
                if not h.watch_proc():
                    continue
                status = h.status()
                if status:
                    # drain the stdout and stderr if the process has exited
                    if h.watch_stdout():
                        h.read_stdout()
                    if h.watch_stderr():
                        h.read_stderr()
                    h.process_cb(status)
                    self.handlers_.remove(h)
        finally:
            self.dtest.lock.release()
        return len(self.handlers_) > 0
        
    def cleanup(self):
        for h in self.handlers_:
            h.kill()
        # wait for the processes to exit...
        while self.wait_for_event(): pass

    def hard_cleanup(self):
        for h in self.handlers_:
            h.kill()
            
#----------------------------------------------------------------------------
class ProcAsyncHandler:
    """Handles asynchronous input/output for a process."""

    READ_BUFFER = 1024

    class SelectStub:
        def __init__(self, proc, fd, cb):
            self.proc_ = proc
            self.fd_ = fd
            self.cb_ = cb

        def fileno(self):
            return self.fd_.fileno()

        def process(self):
            if self.proc_.proc_.poll():
                return
            self.cb_()

    def __init__(self):
        self.dtest_ = None
        self.proc_ = None
        self.stdout_buf_ = ''
        self.stderr_buf_ = ''
    
    def bind(self, proc):
        self.proc_ = proc
        fcntl.fcntl(self.proc_.stdout, fcntl.F_SETFL, os.O_NONBLOCK)
        fcntl.fcntl(self.proc_.stderr, fcntl.F_SETFL, os.O_NONBLOCK)
        
    def pid(self):
        return self.proc_.pid

    def kill(self):
        if not self.proc_:
            return
        
        log.debug('killing %s', self.proc_.pid)
        try:
            os.kill(self.proc_.pid, signal.SIGTERM)
        except OSError:
            pass

    def status(self):
        return self.proc_.poll()

    def stdout(self):
        return ProcAsyncHandler.SelectStub(self,
                                           self.proc_.stdout,
                                           self.read_stdout)

    def stderr(self):
        return ProcAsyncHandler.SelectStub(self,
                                           self.proc_.stderr,
                                           self.read_stderr)

    @staticmethod
    def find_msg_boundary(buf):
        i = buf.find('\n')
        if i == -1:
            return -1
        else:
            return i + 1

    def read_buf(self, fd, bufname):
        while True:
            try:
                s = fd.read(ProcAsyncHandler.READ_BUFFER)
                if s == '':
                    break
                # immutable strings force us to do this silliness
                setattr(self, bufname, getattr(self, bufname) + s)
            except IOError, e:
                if e.errno == errno.EAGAIN:
                    break
        while True:
            i = self.find_msg_boundary(getattr(self, bufname))
            if i == -1:
                break
            self.stdout_cb(getattr(self, bufname)[:i])
            setattr(self, bufname, getattr(self, bufname)[i:])

    def read_stdout(self):
        return self.read_buf(self.proc_.stdout, 'stdout_buf_')

    def read_stderr(self):
        return self.read_buf(self.proc_.stderr, 'stderr_buf_')

    # Override these in subclasses for the correct behavior
    def watch_stdout(self):
        return False

    def watch_stderr(self):
        return False

    def watch_proc(self):
        return True

    def stdout_cb(self, data):
        pass

    def stderr_cb(self, data):
        pass

    def process_cb(self, result):
        pass

#----------------------------------------------------------------------------
class Console(ProcAsyncHandler):
    """Handles user interaction"""

    def stdout(self):
        return ProcAsyncHandler.SelectStub(sys.stdin, self.stdout_cb)
    def pid(self):
        return -1

    def status(self):
        return -1

    def kill(self):
        raise Error("Can't kill console")

    def watch_stdout(self):
        return True

    def stdout_cb(self, data):
        # '' means CRTL-D
        if data[0] == '!':
            exec(data[1:])
        print('User input:"%s"' % data)

    def process_cb(self, event):
        pass    
    
#----------------------------------------------------------------------------
class Host:
    """A physical entity which can receive node allocations."""

    SLEEP_INTERVAL = 0.1                # when waiting, sleep for
                                        # this amount of seconds

    def __init__(self, name, basedir, port_range):
        self.name_ = name
        self.basedir_ = basedir
        self.port_range_ = port_range
        self.next_port_ = port_range[0]
        self.ports_ = {}
        self.interfaces_ = {}           # interface -> name mapping,
                                        # starts out None
        self.dtest_ = None              # set by DTest
        
        # XXX/bowei not quite right b/c of the shared env among hosts
        self.env_frames_ = []

    def name(self):
        return self.name_
        
    def basedir(self):
        return self.basedir_

    def ports(self):
        return self.ports_

    def interfaces(self):
        return self.interfaces_

    def add_avail_interface(self, interface):
        self.interfaces_[interface] = None

    # XXX/bowei - in the future. we probably want to be able to deal
    # with different interfaces as well
    def allocate_port(self, name, port = 0):
        """Allocate a port assigned for use of name. By default a
        random block of ports is chosen.
        @return port assigned to name"""
        if port != 0 and (port < self.port_range_[0] or
                          port > self.port_range_[1]):
            raise ConfigError('Host %s, port %d requested for allocation not in [%d, %d]' %
                              (self.name_, port, self.port_range_[0], self.port_range_[1]))
        elif port == 0:
            if self.next_port_ == self.port_range_[1]:
                raise ConfigError('Host %s, not enough ports, all have been allocated in [%d, %d]' %
                              (self.name_, self.port_range_[0], self.port_range_[1]))
            port = self.next_port_
            self.next_port_ = self.next_port_ + 1
        log.debug('allocated %s:%d to %s' % (self.name_, port, name))
        self.ports_[name] = port
        
        return port

    def allocate_interface(self, name):
        for i in self.interfaces_.keys():
            if self.interfaces_[i] is None:
                self.interfaces_[i] = name
                return i
        raise ConfigError('Host %s ran out of interfaces, trying to assign "%s"' %
                          (self.name_, name))

    def send_file(self, local_file, remote_file):
        """Copy a file to the physical host.
        remote_file will be relative to the basedir_ unless specified
        preceding with a '/'"""
        raise NotImplementedError()

    def recv_file(self, local_file, remote_file):
        """Copy a remote file to local file"""
        raise NotImplementedError()

    def mkdir(self, remote_dir):
        """Create a directory (relative) in the remote
        location. Directory can be nested, all the directories in
        between will be made."""
        raise NotImplementedError()

    def rm_basedir(self):
        """Remove (recursively) all the files in the basedir"""
        raise NotImplementedError()
    
    def run_cmd(self, cmd, timeout = None, env = None,
                cmd_input = None, kill_on_timeout = True):
        """Run a shell command on the host.
        cmd is a list of the command and the arguments
        timeout is in seconds (can be float). negative timeout waits forever.
        cmd_input Additional input to be written to stdin on the client.
        @return tuple (errno, stdout, stderr) of the command"""
        raise NotImplementedError()

    def run_cmd_async(self, cmd, handler):
        """Run a shell command on the asynchronously. Asynchronous
        events are handled by the handler"""
        raise NotImplementedError()

    def run_system(self, cmdstr):
        """Run cmdstr in a shell"""
        raise NotImplementedError()

    def signal(self, pid, signal):
	"""Send a signal to the process pid"""
	raise NotImplementedError()

#----------------------------------------------------------------------------
class LoopbackHost(Host):
    """A host which is implemented as a local loopback.
    port_range is a pair of [min,max] port that can be allocated.
    """

    def __init__(self, name, basedir, port_range):
        Host.__init__(self, name, basedir, port_range)
        self.add_avail_interface('127.0.0.1')
    
    def send_file(self, local_file, remote_file):
        # XXX/bowei -- optimize with hardlinks
        log.debug('send_file "%s" -> "%s"' % (local_file, remote_file))
        dst = os.path.join(self.basedir_, remote_file)
        shutil.copyfile(local_file, dst)
        shutil.copymode(local_file, dst)

    def create_file(self, dest, contents):
	log.debug('create_file "%s"' % dest)
	f = open(os.path.join(self.basedir_, dest), 'w')
	f.write(contents)
	f.close()

    def recv_file(self, remote_file, local_file):
        log.debug('recv_file "%s" -> "%s"' % (remote_file, local_file))
        shutil.copyfile(os.path.join(self.basedir_, remote_file), local_file)


    def mkdir(self, remote_dir):
        try:
            if remote_dir is None:
                log.debug('mkdir basedir "%s"' % self.basedir_)
                os.makedirs(self.basedir_)
            else:
                log.debug('mkdir "%s"' % remote_dir)
                os.makedirs(os.path.join(self.basedir_,remote_dir))
        except OSError:
            pass

    def rm_basedir(self):
        log.debug('rm -r "%s"' % self.basedir_)
        shutil.rmtree(self.basedir_, True)

    def run_cmd_2(self, *cmd):
        self.run_cmd_with_env(None, *cmd)

    def run_cmd_with_env(self, env, *cmd):
        if env:
            self.push_env_(env)
        subprocess.call(list(cmd), cwd = self.basedir_)
        if env:
            self.pop_env_()

    def run_cmd(self, cmd, timeout = None, env = None,
                cmd_input = None, kill_on_timeout = True):
        # XXX/bowei - this logic can be greatly simplified with the
        # new subprocess module
        """Run the specified command, blocking"""
        if cmd[0][0] != '/':
            cmd[0] = os.path.join(self.basedir_, cmd[0])
        log.debug('run_cmd "%s" with timeout=%s' % (self.basedir_, cmd, timeout))
        if env:
            self.push_env_(env)
        proc = subprocess.Popen(cmd,
                                stdin=subprocess.PIPE,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                shell=False,
                                close_fds=True,
                                cwd=self.basedir_)
        if cmd_input:
            proc.stdin.write(cmd_input)
        total_timeout = timeout
        while True:
            errno = proc.poll()
            if errno: break
            common.sleep_sec(Host.SLEEP_INTERVAL)
            if timeout:
                timeout = timeout - Host.SLEEP_INTERVAL
                if timeout <= 0:
                    if env:
                        self.pop_env_()
                    if kill_on_timeout == True:
                        log.debug('run_cmd "%s" timeout kill after %f seconds' %
                                  (cmd, total_timeout))
                        os.kill(proc.pid, signal.SIGKILL)
                    raise Timeout('Timeout running "%s", waited %f seconds' %
                                  (cmd, total_timeout))
        out = proc.stdout.read()
        err = proc.stderr.read()
        log.debug('run_cmd "%s" finished %d' % (cmd, errno))
        if env:
            self.pop_env_()
        return (errno, out, err)
        
    def run_cmd_async(self, cmd, handler, env = None):
        """Run a command asynchronously."""
        if cmd[0][0] != '/':
            cmd[0] = os.path.join(self.basedir_, cmd[0])
        log.debug('run_cmd_async "%s/%s"' % (self.basedir_, cmd))
        if env:
            self.push_env_(env)
        proc = subprocess.Popen(cmd,
                                stdin=subprocess.PIPE,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                shell=False,
                                close_fds=True,
                                cwd=self.basedir_)
        if env:
            self.pop_env_()
        handler.bind(proc)

    def run_system(self, cmdstr):
        os.system(cmdstr)

    def signal(self, pid, signal):
	"""Send a signal to the process pid"""
	os.kill(pid, signal)
        
    def push_env_(self, env):
        """Push the environmental variable mappings onto the environment"""
        frame = {}
        for k in env.keys():
            if k in os.environ:
                frame[k] = os.environ[k]
            else:
                frame[k] = None
        for k in env.keys():
            os.environ[k] = str(env[k])
        self.env_frames_.append(frame)

    def pop_env_(self):
        """Pop the environmental variable mappings from the environment"""
        frame = self.env_frames_.pop()
        for k in frame.keys():
            if frame[k] is None:
                del(os.environ[k])
            else:
                os.environ[k] = frame[k]

#----------------------------------------------------------------------------
class RemoteHost(Host):
    """A host that is on a remote machine"""
    # XXX/bowei
    pass

#----------------------------------------------------------------------------
class Node:
    """Each test run consists of N node, each running some
    commands. Nodes are virtual entities which are mapped by the test
    system to Hosts, which represent real allocation of machines,
    network ports, etc."""

    class SymPort:
        """This represents a symbolic port allocation that will be
        instantiated by the mapper. After the mapper has done the
        basic mapping of the ports, then you can use this class to
        retrieve the actually addr:port mapping."""
        def __init__(self, node, iface_name, port_name):
            self.node = node
            self.iface_name = iface_name
            self.port_name = port_name
            
        def ap(self):
            return (self.node.interfaces_[self.iface_name],
                    self.node.ports_[self.port_name])

    def __init__(self, name):
        self.name_ = name
        self.dirs_ = []                 # list of dirs to create
        self.manifest_ = {}             # dict of local -> remote files
	self.dyn_files_ = {}            # remote name -> content
        self.ports_ = {}                # dict of name -> port assignments
        self.interfaces_ = {}
        
        self.host_ = None               # assigned by the mapper
        self.dtest_ = None              # set by DTest
        
    def name(self):
        return self.name_

    def host(self):
        return self.host_

    def ports(self):
        return self.ports_

    def interfaces(self):
        return self.interfaces_

    def port(self, p):
        return self.ports_[p]

    def interface(self, i):
        return self.interfaces_[i]

    @staticmethod
    def test_method(f):
        """Decorator to ensure that we are thread-safe from the test
        thread"""
        def wrapper(self, *args):
            self.dtest_.lock.acquire()
            try:
                return f(self, *args)
            finally:
                self.dtest_.lock.release()
        return wrapper

    def reg_php_package(self, package):
        log.debug('Adding files from python package "%s"' % package)
        if not (package in self.dirs_):
            self.dirs_.append(package)
            log.debug('Adding dir %s for PHP package' % package)
        for php in glob.glob('./%s/*.php' % package):
            self.manifest_[php] = php
            log.debug('Adding file %s for PHP package' % php)

    def reg_py_package(self, package):
        """Add all the python files in a package to the manifest"""
        log.debug('Adding files from python package "%s"' % package)
        path = package.replace('.','/')
        if not (path in self.dirs_):
            self.dirs_.append(path)
            log.debug('Adding dir %s for python package' % path)
        for py in glob.glob('./%s/*.py' % path):
            self.manifest_[py] = py
            log.debug('Adding file %s for python package' % py)

    def reg_file(self, file, dest = None):
        """Add a file to the manifest"""
        if dest is None:
            dest = file
        log.debug('Adding file %s -> %s to the manifest' % (file, dest))
        self.manifest_[file] = dest
        if (os.path.dirname(dest) not in self.dirs_) and os.path.dirname(dest) != '':
            self.dirs_.append(os.path.dirname(dest))
            log.debug('Adding dir %s for file %s' % (os.path.dirname(dest), file))

    def reg_dynfile(self, dest, contents):
	"""Add a dynamic file, generated by the test script"""
	log.debug('Adding dynamic file %s to the manifest' % dest)
	self.dyn_files_[dest] = contents
        if (os.path.dirname(dest) not in self.dirs_) and os.path.dirname(dest) != '':
            self.dirs_.append(os.path.dirname(dest))
            log.debug('Adding dir %s for dynfile' % os.path.dirname(dest))

    def reg_interface(self, name):
        log.debug('Allocating an interface "%s" to %s' % (name, self.name_))
        self.interfaces_[name] = None
        
    def reg_port(self, name, interface = None, assigned = None):
        if interface is None:
            interface = 'main'
        if assigned is None:
            assigned = 0
        self.ports_[name] = assigned
        log.debug('Adding port %s:%d to %s' % (name, assigned, self.name_))

        return Node.SymPort(self, interface, name)

    def set_host(self, host):
        self.host_ = host

    def setup_logging(self):
        """Setup and register appropriate logging for this node"""
        pass
            
    def setup_files(self):
        log.debug('setting up files for node %s on host %s' %
                  (self.name_, self.host_.name()))
        self.host_.mkdir(None)
        for d in self.dirs_:
            self.host_.mkdir(d)
        for local in self.manifest_.keys():
            self.host_.send_file(local, self.manifest_[local])
	for dynfile in self.dyn_files_.keys():
	    self.host_.create_file(dynfile, self.dyn_files_[dynfile])

    def cleanup_files(self):
        self.host_.rm_basedir()
        # XXX/bowei -- perhaps make this more precise

    def cleanup(self):
        self.cleanup_files()

#----------------------------------------------------------------------------
class Mapper:
    """Maps nodes to actual hosts."""
    
    def __init__(self):
        self.dtest_ = None              # set by DTest

    def map(self, nodelist):
        raise NotImplementedError()

#----------------------------------------------------------------------------
class LocalMapper:
    """Maps to run in the standard local loopback way."""

    def __init__(self, basedir = None, max_ports = 100, base_port = 9000):
        self.max_ports_ = max_ports
        self.base_port_ = base_port
        self.basedir_ = basedir         # set by Dtest if None
        
    def get_host(self, node_name):
        id = len(self.dtest_.hosts)
        name = '%d-%s' % (id, node_name)
        h = LoopbackHost(name,
                         os.path.join(self.basedir_, name),
                         (self.base_port_ + id * self.max_ports_,
                         self.base_port_ + (id + 1) * self.max_ports_))
        self.dtest_.add_host(h)
        return h

    def map(self, nodelist):
        for n in nodelist:
            n.set_host(self.get_host(n.name()))

            # assign interfaces and ports
            port_names = n.ports_.keys()
            port_names.sort()
            for p in port_names:
                n.ports_[p] = n.host_.allocate_port(p, n.ports_[p])
                log.debug('mapped %s:%s to %s' %
                          (n.name(), p, n.ports_[p]))
            for i in n.interfaces_.keys():
                n.interfaces_[i] = n.host_.allocate_interface(i)

        # XXX/bowei -- don't need to do anything for now...
        # XXX/bowei -- link up all of the symbolic ports
        for n in nodelist:
            pass
            
#----------------------------------------------------------------------------
class Logger:
    """Handles the logging from the run programs"""

    # XXX/bowei -- maybe use the Python logging system?
    def __init__(self):
        self.rules_ = []
        self.sinks_ = {}
        self.add_sink(LogSink('null'))
        self.dtest_ = None              # set by DTest
        
    def log(self, path, msg):
        sink = self.match_sink(path)
        sink.log(path, msg)

    def add_rule(self, rule, sink = None):
        if sink is None:
            sink = rule
        self.rules_.append((rule, sink))

    def add_sink(self, sink):
        if sink.name_ in self.sinks_:
            raise ConfigError('Multiple LogSinks with the same name "%s"' %
                              sink.name_)
        self.sinks_[sink.name_] = sink

    def get_sink(self, name):
        return self.sinks_[name]
        
    def match_sink(self, path):
        # XXX/bowei stub
        for r, s in self.rules_:
            if r == path:
                return self.sinks_[s]
        return self.sinks_['null']

#----------------------------------------------------------------------------
class DefAsyncHandler(ProcAsyncHandler):
    """Generate log messages from an asynchronous proc source"""
    def __init__(self, path, default = True):
        ProcAsyncHandler.__init__(self)
        self.path_  = path
        self.dtest_ = None              # set by DTest

        self.stdout_cb_ = []
        self.stderr_cb_ = []

        if default:
            self.add_stdout_cb(self.default_cb)
            self.add_stderr_cb(self.default_cb)

    def default_cb(self, path, data):
        self.dtest_.logger.log(path, data)

    def add_stdout_cb(self, cb):
        self.stdout_cb_.append(cb)

    def add_stderr_cb(self, cb):
        self.stderr_cb_.append(cb)

    # virtual from ProcAsyncHandler
    def watch_stdout(self):
        return True

    def watch_stderr(self):
        return True

    def watch_proc(self):
        return True

    def stdout_cb(self, data):
        for cb in self.stdout_cb_:
            cb(self.path_, data)

    def stderr_cb(self, data):
        for cb in self.stderr_cb_:
            cb(self.path_, data)

    # XXX/bowei -- what about process_cb?
    def process_cb(self, result):
        self.dtest_.logger.log(self.path_, "process exit %s" % result)

#----------------------------------------------------------------------------
class LogSink:
    def __init__(self, name):
        self.name_ = name
        self.next_ = None

    def log(self, path, msg):
        if self.next_:
            self.next_.log(path, msg)

    def chain(self, next):
        self.next_ = next

    def format(self, path, msg):
        return '%s:\t%s' % (path, msg)

#----------------------------------------------------------------------------
class PrintLogSink(LogSink):
    def __init__(self, name):
        LogSink.__init__(self, name)

    def log(self, path, msg):
        LogSink.log(self, path, msg)
        s = self.format(path, msg)
        print(s.strip())

#----------------------------------------------------------------------------
class FileLogSink(LogSink):
    def __init__(self, name, filename = None):
        if filename is None:
            filename = name + '.log'
        LogSink.__init__(self, name)
        # XXX/bowei -- open without buffering
        self.f_ = open(filename, 'a')

    def log(self, path, msg):
        LogSink.log(self, path, msg)
        s = self.format(path, msg)
        self.f_.writeline(s.strip())

#----------------------------------------------------------------------------
class Watch:
    def __init__(self, pred, pargs, action, aargs):
        self.pred = pred
        self.pargs = pargs
        self.action = action
        self.aargs = aargs

    def __call__(self):
        if self.pred(*self.pargs):
            self.action(*self.aargs)
            return False
        return True

#----------------------------------------------------------------------------
class TestRunner:
    """This encapsulates the thread that will run the test script"""

    POLL_INTERVAL = 0.01

    def __init__(self, dtest, f, *args):
        self.dtest = dtest
        self.f = f
        self.args = args
        self.thread = threading.Thread(target = self.run)

    def run(self):
        try:
            self.f(self, *self.args)
        finally:
            self.end()

    def start(self):
        self.thread.start()

    def end(self):
        self.exec_f(self.dtest.set_quit)

    def lock(self):
        self.dtest.lock.acquire()

    def unlock(self):
        self.dtest.lock.release()

    def sleep(self, timeout):
        start = time.time()
        while True:
            self.lock()
            try:
                if self.dtest.quit:
                    return
            finally:
                self.unlock()
            common.sleep_sec(TestRunner.POLL_INTERVAL)
            if time.time() - start > timeout:
                return

    def pause(self):
	print('-- paused, hit <return> to continue --')
	sys.stdin.readline()

    def wait_for(self, timeout, predicate, *args):
        """Polling implementation is not exactly the best..."""
        start = time.time()
        while True:
            self.lock()
            try:
                if self.dtest.quit:
                    return
                p = predicate(*args)
            finally:
                self.unlock()
            if p:
                break
            common.sleep_sec(TestRunner.POLL_INTERVAL)
            if timeout and time.time() - start > timeout:
                raise Timeout()

    def exec_f(self, f, *args):
        """This function is to wrap code so that the right lock is
        used. Sigh. I wish python had some kind of macro ability"""
        
        self.lock()
        try:
            f(*args)
        finally:
            self.unlock()

    def exec_py(self, str, l, g):
        """This function is to wrap code so that the right lock is
        used. Sigh. I wish python had some kind of macro ability"""
        self.lock()
        try:
            exec str in l, g
        finally:
            self.unlock()
